package com.googlecode.jlogalizer.mapred.log;

import java.io.File;
import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.googlecode.jlogalizer.mapred.log.map.LogMapper;
import com.googlecode.jlogalizer.mapred.log.reduce.LogReducer;
import com.googlecode.jlogalizer.mapred.util.MapReduceFileUtils;

public class LogMapReduceJob extends Configured implements Tool{

	private static void initJob(String jobName, Configuration config, String inputPath, String outputPath) throws IOException, InterruptedException, ClassNotFoundException{
	
		MapReduceFileUtils.deleteDirectory(new File(outputPath));
		Job job=new Job(config, jobName);
		job.setJarByClass(LogMapReduceJob.class);
		job.setMapperClass(LogMapper.class);
		job.setReducerClass(LogReducer.class);
		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		FileInputFormat.addInputPath(job, new Path(inputPath));
		FileOutputFormat.setOutputPath(job, new Path(outputPath));
		
		job.waitForCompletion(true);
		
		     
		
	}
	
	public static void main(String[] args) throws Exception {
		ToolRunner.run(new LogMapReduceJob(), args);
	}

	@Override
	public int run(String[] args) throws Exception {
		
		Configuration config= getConf();
		System.out.println("Args [0] :"+args[0]);
		System.out.println("Args [1] :"+args[1]);
		System.out.println("Args [2] :"+args[2]);
		/*config.set("log_file_type", "apache_access_log");
		config.setInt("position_in_log_file", 1);*/
		
		initJob(args[0], config, args[1], args[2]);
		return 0;
	}
	
	
}
